package com.cbh.amadeus.client.receiver.consumer;

import com.cbh.amadeus.client.annotation.AmadeusMessageRegister;
import com.cbh.amadeus.client.domain.MessageResultModel;
import com.cbh.amadeus.client.receiver.service.MessageHolder;
import com.cbh.amadeus.client.receiver.tool.MessageResultModelTool;
import com.cbh.amadeus.client.utils.ChannelReference;
import com.cbh.amadeus.client.utils.MessageHolderProxyChange;
import com.cbh.amadeus.common.enums.MessageResult;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.utils.StringUtil;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.AnnotationUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author ChenBiHui
 * @date 2025/4/28
 */
@Slf4j
public class MessageConsumerProxy {

    private final MessageResultModelTool tool;
    private final MessageHolderWrap holder;
    private final Map<String, List<MessageHolder<?>>> topicWithHolders = new HashMap<>(64);

    public MessageConsumerProxy(MessageResultModelTool tool, MessageHolderWrap holder, List<MessageHolder<?>> holders) {
        this.tool = tool;
        this.holder = holder;
        this.topicWithHolders.putAll(
                holders.stream()
                        .collect(Collectors.groupingBy(item -> {
                            AmadeusMessageRegister register = AnnotationUtils.findAnnotation(item.getClass(), AmadeusMessageRegister.class);
                            if (register == null || StringUtil.unHasText(register.topic())) {
                                throw new IllegalArgumentException("@MessageRegister not found or @MessageRegister#topic() is blank");
                            } else {
                                return register.topic();
                            }
                        }))
        );
    }

    public void receive(Channel channel, AmadeusMessage message) {
        List<MessageHolder<?>> holders = this.topicWithHolders.get(message.getTopic());
        if (holders == null || holders.isEmpty()) {
            log.warn(
                    "===== EVENT [GEN] Topic [{}] can't find any [MessageHolder] =====",
                    message.getTopic()
            );
            return;
        }
        String name = ChannelReference.findName(channel);
        holder.match(channel, name, message, holders);
    }

    public void resend(Channel channel, AmadeusMessage message) {
        List<MessageHolder<?>> holders = this.topicWithHolders.get(message.getTopic());
        if (holders == null || holders.isEmpty()) {
            log.warn(
                    "===== EVENT [RESEND] Topic [{}] can't find any [MessageHolder] =====",
                    message.getTopic()
            );
            return;
        }
        String name = ChannelReference.findName(channel);
        List<MessageResultModel> complete = tool.complete(message.getTenant(), message.getTopic(), message.getMessageId(), name);
        if (!(complete == null || complete.isEmpty())) {
            Set<String> labels = complete.stream()
                    .filter(item -> MessageResult.SUCCESS.name().equals(item.getInvokeResult()))
                    .map(MessageResultModel::getLabel)
                    .collect(Collectors.toSet());
            holders = holders.stream()
                    .filter(item -> !labels.contains(MessageHolderProxyChange.get(item)))
                    .collect(Collectors.toList());
            if (holders.isEmpty()) {
                ChannelReference.ack(channel, message, name);
                return;
            }
        }
        holder.match(channel, name, message, holders);
    }
}
